package cn._51doit.live.utils;


import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;

import java.util.Arrays;
import java.util.List;
import java.util.Properties;

public class FlinkUtils {

    public static final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    public static <T> DataStream<T> createKafkaStream(ParameterTool parameterTool, Class<? extends DeserializationSchema<T>> deserializationSchema) throws IllegalAccessException, InstantiationException {
        Long checkpointInterval = parameterTool.getLong("checkpoint.interval", 120000);
        //开启checkpoint
        env.enableCheckpointing(checkpointInterval);

        String inputTopicsStr = parameterTool.getRequired("kafka.input.topics");
        List<String> topicList = Arrays.asList(inputTopicsStr.split(","));
        String bootstrapServers = parameterTool.getRequired("bootstrap.servers");
        String autoOffsetReset = parameterTool.get("auto.offset.reset", "earliest");
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", bootstrapServers);
        properties.setProperty("auto.offset.reset", autoOffsetReset);


        FlinkKafkaConsumer<T> kafkaConsumer = new FlinkKafkaConsumer<T>(
                topicList,
                deserializationSchema.newInstance(),
                properties
        );
        kafkaConsumer.setCommitOffsetsOnCheckpoints(false); //将偏移量只保存在状态中


        return env.addSource(kafkaConsumer);
    }

    public static <T> DataStream<T> createKafkaStreamWithId(ParameterTool parameterTool, KafkaDeserializationSchema<T> deserializationSchema) {

        Long checkpointInterval = parameterTool.getLong("checkpoint.interval", 120000);
        //开启checkpoint
        env.enableCheckpointing(checkpointInterval);

        String inputTopicsStr = parameterTool.getRequired("kafka.input.topics");
        List<String> topicList = Arrays.asList(inputTopicsStr.split(","));
        String bootstrapServers = parameterTool.getRequired("bootstrap.servers");
        String autoOffsetReset = parameterTool.get("auto.offset.reset", "earliest");
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", bootstrapServers);
        properties.setProperty("auto.offset.reset", autoOffsetReset);


        FlinkKafkaConsumer<T> kafkaConsumer = new FlinkKafkaConsumer<T>(
                topicList,
                deserializationSchema,
                properties
        );
        kafkaConsumer.setCommitOffsetsOnCheckpoints(false); //将偏移量只保存在状态中


        return env.addSource(kafkaConsumer);

    }
}
